package org.qy.star_server.core;

import io.grpc.Server;
import io.grpc.ServerBuilder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.qy.StarServiceApplication;
import org.qy.star_common.enums.ExchangeType;
import org.qy.star_common.message.Message;
import org.qy.star_common.util.Constants;
import org.qy.star_common.util.PathUtil;
import org.qy.star_server.data.center.DiskDataCenter;
import org.qy.star_server.data.center.MemoryDataCenter;
import org.qy.star_server.data.mapper.MetaMapper;
import org.qy.star_server.grpc.BrokerService;

import java.io.*;
import java.util.*;

/**
 * @author yinger
 * @description TODO
 * @date 2024/3/30 13:53
 **/
@Slf4j
@Data
public class BrokerServer {
    public static DiskDataCenter disk;
    public static MemoryDataCenter memory = MemoryDataCenter.getMemoryBean();
    private volatile boolean isInterrupt = true;    // 中断服务器程序
    private int port;

    public BrokerServer() {
    }

    private BrokerServer(int port) {
        this.port = port;
    }

    // 启动服务
    public void start() {
        log.info("{} 端口服务启动中 ... ", port);
        // 绑定服务到 7993 端口
        ServerBuilder<?> serverBuilder = ServerBuilder.forPort(port);
        // 发布服务
        serverBuilder.addService(new BrokerService());
        // 创建服务对象
        Server server = serverBuilder.build();
        // 启动服务
        try {
            server.start();
            log.info("{} 服务器启动成功 ... ", port);
            // 阻塞主线程 不断提供服务
            server.awaitTermination();
            // 此处接收 Socket 请求

        } catch (IOException | InterruptedException e) {
            throw new RuntimeException(e);
        }
        log.info("服务器关闭 ... ");
    }


    public static BrokerServer brokerServerFactory(int port) {
        try {
            disk = new DiskDataCenter(StarServiceApplication.context.getBean(MetaMapper.class));
            BrokerServer brokerServer = new BrokerServer(port);
            brokerServer.init();
            return brokerServer;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void init() throws IOException {
        log.info("init broker server ... ");
        PathUtil.createBrokerDir();
        if (!checkDBExists()) {
            log.info("init database ... ");
            File file = new File("./data");
            if (!file.exists()) {
                file.mkdirs();
            }
            File dbFile = new File("./data/meta.db");
            if (!dbFile.exists()) {
                dbFile.createNewFile();
            }
            disk.createTables();
            insertDefaultData();
        }
        // 这个方法调用了，说明服务重启了，需要恢复内存数据
        recoveryData();
    }

    public boolean checkDBExists() {
        File file = new File("./data/meta.db");
        return file.exists();
    }

    public void insertDefaultData() {
        User adminUser = new User(Constants.DEFAULT_USER_USERNAME, Constants.DEFAULT_USER_PASSWORD);
        disk.insertUser(adminUser.getUsername(), adminUser.getPassword());
        memory.insertUser(adminUser.getUsername(), adminUser.getPassword());
        VirtualHost virtualHost = new VirtualHost(Constants.DEFAULT_USER_USERNAME, Constants.DEFAULT_VIRTUAL_HOST_NAME);
        disk.insertVirtualHost(virtualHost.getUsername(), virtualHost.getVirtualHostName());
        memory.insertVirtualHost(virtualHost.getUsername(), virtualHost.getVirtualHostName());
        Exchange exchange = new Exchange(Constants.DEFAULT_USER_USERNAME, Constants.DEFAULT_VIRTUAL_HOST_NAME, Constants.DEFAULT_EXCHANGE_NAME, ExchangeType.DIRECT, true, false, null);
        disk.insertExchange(exchange.getUsername(), exchange.getVirtualHostName(), exchange.getExchangeName(), exchange.getExchangeType(), exchange.isDurable(), exchange.isAutoDelete(), null);
        memory.insertExchange(exchange.getUsername(), exchange.getVirtualHostName(), exchange.getExchangeName(), exchange.getExchangeType(), exchange.isDurable(), exchange.isAutoDelete());
    }

    public void recoveryData() {
        try {
            // 恢复用户数据
            List<User> users = disk.selectAllUser();
            for (User user : users) {
                memory.insertUser(user.getUsername(), user.getPassword());
            }

            // 恢复虚拟主机数据
            List<VirtualHost> virtualHosts = disk.selectAllVirtualHost();
            for (VirtualHost virtualHost : virtualHosts) {
                memory.insertVirtualHost(virtualHost.getUsername(), virtualHost.getVirtualHostName());
            }

            // 恢复交换机数据
            List<Exchange> exchanges = disk.selectAllExchange();
            for (Exchange exchange : exchanges) {
                memory.insertExchange(exchange.getUsername(), exchange.getVirtualHostName(), exchange.getExchangeName(), exchange.getExchangeType(), exchange.isDurable(), exchange.isAutoDelete());
            }

            // 恢复队列数据  恢复完队列之后消息统计文件默认值为 0,0
            List<org.qy.star_server.core.Queue> queues = disk.selectAllQueue();
            for (org.qy.star_server.core.Queue queue : queues) {
                memory.insertQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete());
            }

            // 恢复绑定数据
            List<Binding> bindings = disk.selectAllBinding();
            for (Binding binding : bindings) {
                memory.insertBinding(binding.getUsername(), binding.getVirtualHostName(), binding.getExchangeName(), binding.getQueueName(), binding.getBindingKey());
            }

            // 恢复消息数据（待确认消息只存放在内 存，无法恢复）
            // 恢复所有队列的消息数据
            int i = 0;
            for (org.qy.star_server.core.Queue queue : queues) {
                if (queue != null && queue.getQueueName() != null) {
                    memory.insertQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete());
                    LinkedList<Message> messages = disk.getFileManager().loadValidMessages(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
                    // 重启时加载有效消息
                    Queue q = memory.selectQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName());
                    q.setFileTotalCount(messages.size());
                    q.setFileValidCount(messages.size());
                    for (Message message : messages) {
                        memory.insertMessageToCenter(message);
                        memory.insertMessageToQueue(queue.getUsername(), queue.getVirtualHostName(), queue.getQueueName(), message);
                        VirtualHost virtualHost = memory.selectVirtualHost(queue.getUsername(), queue.getVirtualHostName());
                        Token token = new Token(queue.getUsername(), virtualHost.getVirtualHostName(), queue.getQueueName());
                        virtualHost.getTokenQueue().put(token);
                        i++;
                    }
                }
            }
            log.info("recovery " + i + " messages to memory !");
        } catch (InterruptedException e) {
            log.error("恢复数据到内存失败");
            e.printStackTrace();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}